package com.lvmama.rhino.common.utils.methodset


import org.apache.spark.sql.{Column, DataFrame}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

/**
  * Created by yuanxiaofeng on 2016/10/19.
  */
object CalculateFunction{

  val windowSpec = (param1: Column, param2: Column) => Window.partitionBy(param1).orderBy(param2.asc)

  val windowSpec_desc = (param1: Column, param2: Column) => Window.partitionBy(param1).orderBy(param2.desc)

  val windowRange = (param1: Column, param2: Column) => Window.partitionBy(param1).orderBy(param2)
    .rangeBetween(-30 * 60 * 1000l, 0l)

  val columnLead = (param: Column) => lead(param, 1).over(windowSpec(col("sessionId"), col("timestamp")))
  /**
    * 将参数所在的列往后延迟一行
    */
  val columnLag = (param: Column) => lag(param, 1).over(windowSpec(col("sessionId"), col("timestamp")))

  /**
    * 创建 productId column
    */
  val createProductIdCol = udf((pageType: Map[String, String]) => {
    pageType.get("productId").get
  })


  /**
    * 自定义函数，用来新增列
    * 新列为param2-param1的值
    */
  val column_delta = udf((param1: Long, param2: Long) => {
    param2 - param1
  })

  /**
    * 自定义函数，用来新增列
    * 新列为param1 +　param2
    */
  val column_sum = udf((param1: Long, param2: Long) => {
    param2 + param1
  })

  val value_range_Interval = udf((col: Long, base: Long) => {
    ((col % base) * base /1000 + 1, (col % base + 1) * base /1000)
  })

  val value_range = udf((col: Long, base: Long) => {
    col / base + 1
  })

  var static_uuid: String = uuid()

  def uuid() = java.util.UUID.randomUUID.toString.replace("-", "")

  /**
    * warning: 在调用此方法时需要设置 executor-cores 为 1，
    * 不然达不到预期效果。
    * @return
    */
  def traversalAssign = udf((token: String, num: Int) => {
    var id: String = ""
    if (num == 1) {
      static_uuid = uuid() + "_" + token
      id = static_uuid
    } else {
      id = static_uuid
    }
    id
  })

  /**
    * 字段校验
    */
  val fieldVerify = (fields: DataFrame) => {
    fields
      .filter(length(col("logType")) <= 20)
      .filter(length(col("deviceToken")) <= 50)
      .filter(length(col("operator")) <= 20)
      .filter(length(col("sessionId")) <= 100)
      .filter(length(col("longitude")) <= 20)
      .filter(length(col("latitude")) <= 20)
      .filter(length(col("province")) <= 20)
      .filter(length(col("userSource")) <= 20)
      .filter(length(col("userId")) <= 20)
      .filter(length(col("buttonCode")) <= 20)
      .filter(length(col("pageCode")) <= 50)
      .filter(length(col("channelCode")) <= 50)
      .filter(length(col("platformCode")) <= 20)
      .filter(length(col("pageTypeCode")) <= 20)
      .filter(length(col("timestamp")) <= 20)
      .filter(length(col("stationId")) <= 20)
      .filter(length(col("holeCode")) <= 20)
      .filter(length(col("accessCode")) <= 50)
      .filter(length(col("newVisitor")) <= 20)

    fields
  }
}

